
#ifndef __SDK__HAMAL_BASE_H__INCLUDE__
#define __SDK__HAMAL_BASE_H__INCLUDE__

#include "ServiceBase.h"
#include "BuiltinTask.h"
#include "Utility/Mutex.h"
#include "Utility/TaskScheduler.h"
#include <map>
#include <set>
#include <string>
#include <signal.h>


namespace nebula { namespace sdk {

class HamalBase : public ServiceBase
{
public:
	HamalBase();

	typedef PlainTextTask HamalTask; // TODO: consume too many memory for large files; implement and use LargeTextTask

public:
	virtual NEBULA_ERROR_CODE OnInitialize  () = 0;
	virtual NEBULA_ERROR_CODE OnStart       () = 0;
	virtual void              OnStop        () = 0;
	virtual void              OnFinalize    () = 0;

protected:
	// initialize the hamal
	NEBULA_ERROR_CODE InitializeHamal       (const char* config_file);

	// finalize the hamal
	void              FinalizeHamal         ();

	// start the hamal
	NEBULA_ERROR_CODE StartHamal            ();

	// stop the hamal
	void              StopHamal             ();

	// start the external worker; set working_dir to NULL if the target
	// working directory is current one, other arguments are similar to
	// execve(2). TODO: a watch dog for external worker is needed
	NEBULA_ERROR_CODE StartExternalWorker   (const char* working_dir,
	                                         const char* path,
	                                         char* const argv[],
	                                         char* const envp[]);

	// stop the external worker
	NEBULA_ERROR_CODE StopExternalWorker    (int sig=SIGINT, unsigned timeout=5000);

protected:
	virtual void      OnAnteSerializeTask   (ITask*);
	virtual void      OnPostUnSerializeTask (ITask*);

private:
	NEBULA_ERROR_CODE LoadSlotsInfoFromFile (const char* file);

private:
	struct slot_info_t
	{
		std::string slot_name;
		bool is_push_slot; // true for push slot, false for pop slot
		std::string meta_extension;
		std::string data_extension;
		std::string path;
		bool meta_only;
	};

	typedef std::map<std::string, slot_info_t> slots_info_t;
	slots_info_t m_slots;

	typedef std::map<std::string, ITodoTaskQueue*> slot_iqueue_map_t;
	typedef std::map<std::string, IDoneTaskQueue*> slot_oqueue_map_t;
	slot_iqueue_map_t m_iqueues;
	slot_oqueue_map_t m_oqueues;

	std::string m_external_worker;
	pid_t m_external_worker_pid;

	typedef std::map<std::string, std::set<std::string> > doing_tasks_map_t;
	doing_tasks_map_t m_doings;
	utility::Mutex m_doings_lock;

private:
	typedef utility::TaskScheduler<HamalBase> hamal_scheduler_type;

	hamal_scheduler_type m_incoming_scheduler;
	NEBULA_ERROR_CODE process_incoming(bool &repeat_immediately);
	NEBULA_ERROR_CODE process_incoming_files(
		ITodoTaskQueue &queue, const slot_info_t &slot_info, unsigned int &processed);

	hamal_scheduler_type m_outgoing_scheduler;
	NEBULA_ERROR_CODE process_outgoing(bool &repeat_immediately);
	NEBULA_ERROR_CODE process_outgoing_files(
		IDoneTaskQueue &queue, const slot_info_t &slot_info, unsigned int &processed);

	hamal_scheduler_type m_doing_checker;
	NEBULA_ERROR_CODE check_doing_tasks(bool &repeat_immediately);
	NEBULA_ERROR_CODE check_doing_tasks(const slot_info_t &);
};

}} //end of nebula::sdk

#endif // __SDK__HAMAL_BASE_H__INCLUDE__

